package main

import (
	"fmt"
	"github.com/MingOf/logDemo/conf"
	"github.com/MingOf/logDemo/kafka"
	"github.com/Shopify/sarama"
	"logtransfer/es"
	"sync"
)

/*
log transfer

将日志数据从 kafka 中取出来发往 es
*/
func main() {
	var wg sync.WaitGroup
	var (
		err      error
		consumer sarama.Consumer
		e        *es.Es
	)
	//0.加载配置
	cfg := conf.NewConfig("./conf/config.ini")
	//1.初始化
	fmt.Println(cfg)

	// 2.新建 es 客户端
	e = es.NewEsClient(cfg.Es.Address)
	e.CurrentIndex = cfg.Kafka.Topic

	// 3.新建 kafka 消费者
	if consumer, err = kafka.NewConsumer([]string{cfg.Kafka.Address}); err != nil {
		fmt.Println("cannot generate consumer,err:", err)
		return
	}

	// 4.从 kafka 中取出消息
	wg.Add(1)
	if err = kafka.GetFromKafka(consumer, "web-log", e.SendToEs); err != nil {
		fmt.Println("cannot get from kafka,err:", err)
		return
	}
	wg.Wait()
}
